package cn.gupao.dao;

import cn.gupao.pojo.LogBean;
import cn.gupao.utils.ConnectionUtils;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;
import java.util.Set;

public class ProfileHbaseDaoImpl implements ProfileDao {

    private Connection hbaseConn;

    public void init(ParameterTool parameterTool) throws Exception{
        hbaseConn = ConnectionUtils.getHbaseConn(parameterTool);
    }

    @Override
    public boolean isMatchProfile(LogBean bean, Map<String, String> profileCondition) throws Exception {

        //查询Hbase，与用户实际的画像进行对比
        if (profileCondition != null && profileCondition.size() > 0) {
            Table table = hbaseConn.getTable(TableName.valueOf("eagle_profile"));
            //获取当前用户ID
            Get get = new Get(Bytes.toBytes(bean.getDeviceId()));
            //添加查询条件
            Set<String> tags = profileCondition.keySet();
            for (String tag : tags) {
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(tag));
            }
            Result result = table.get(get);
            table.close();
            if (result != null) {
                for (String tag : tags) {
                    //获取hbase中实际的画像value
                    byte[] value = result.getValue(Bytes.toBytes("f"), Bytes.toBytes(tag));
                    //获取事先设置的画像value
                    String tagValue = profileCondition.get(tag);
                    if (value != null) {
                        if (!tagValue.equals(new String(value))) {
                            return false;
                        }
                    } else {
                        return false;
                    }
                }
            } else {
                return false;
            }
        }
        return true;
    }

    @Override
    public void close() throws Exception {
        hbaseConn.close();
    }
}
